use crate::codec::d_codec::{DCodecError, Packet, PacketBytesCodec, Transmission};
use crate::serializer::serialization::Serialized;
use crate::start::dispatcher::PacketDispatcher;
use crate::start::server::{Message, Server, Service, Shared};
use crate::start::{Connection, SendMsg};
use async_trait::async_trait;
use bytes::{BufMut, BytesMut};
use futures::stream::SplitStream;
use futures::{SinkExt, StreamExt};
use futures_core::stream::Stream;
use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::{io, thread};
use std::error::Error;
use futures::future::ok;
use tokio::net::UdpSocket;
use tokio::runtime::Runtime;
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use tokio::sync::{mpsc, Mutex};
use tokio_util::codec::{BytesCodec, Decoder};
use tokio_util::udp::UdpFramed;
use crate::start::processor::Processor;

pub struct UdpServer;

#[async_trait]
impl Server for UdpServer {
    type Socket = UdpSocket;

    async fn bind(address: SocketAddr) -> io::Result<Self::Socket> {
        Ok(UdpSocket::bind(address).await?)
    }

    async fn poll(socket: Self::Socket, service: Arc<PacketDispatcher>) {
        let (send, mut recv) = mpsc::unbounded_channel();
        let mut processor = UdpProcessor {
            send: send,
            recv: Some(recv),
        };
        processor.poll(socket, service).await;
        // let state = Arc::new(Mutex::new(Shared::new()));
        // // let mut framed = UdpFramed::new(socket, BytesCodec::new());
        // let framed = UdpFramed::new(socket, PacketBytesCodec::default());
        // let codec = *framed.codec();
        // let (mut sink, mut stream) = framed.split();
        // let (send, mut recv) = mpsc::unbounded_channel();
        // // upd没有链接所以只有一个UdpConnection，所以这里分担分发工作
        // runtime.spawn(async move {
        //     while let x1 = recv.recv().await.unwrap_or(Message::Close()) {
        //         match x1 {
        //             Message::Send(packet, socket_addr) => {
        //                 sink.send((packet, socket_addr)).await;
        //                 sink.flush().await;
        //                 // sink.start_send((Packet,SocketAddr));
        //             }
        //             Message::Broadcast(_, _) => {}
        //             Message::Forward(_, _) => {}
        //             Message::Received(_, _) => {}
        //             Message::Close() => {}
        //         }
        //     }
        // });
        // let mut udp_connection = UdpConnection::new(stream, state.clone(), send.clone()).unwrap();
        //
        // while let Some(result) = udp_connection.next().await {
        //     match result {
        //         // A message was received from a peer. Send it to the
        //         // current user.
        //         Ok(Message::Received(mut packet, addr)) => {
        //             // peer.lines.send(&msg).await?;
        //             let send = send.clone();
        //             let service = service.clone();
        //             //   todo  Dispatcher
        //             runtime.spawn(async move {
        //                 println!(
        //                     "收到新链接:{:?},message:{:?},线程:{:?}",
        //                     addr,
        //                     packet,
        //                     thread::current().id()
        //                 );
        //
        //                 let connection = Connection {
        //                     packet: packet,
        //                     send: send,
        //                     addr,
        //                 };
        //                 println!("收到新链接{:?}", connection);
        //                 let service = service.clone();
        //                 tokio::spawn(async move {
        //                     //     Dispatcher
        //                     service.dispatcher(connection);
        //                 });
        //                 // UdpServer::process(udp_connection,packet,addr).await;
        //                 // println!("sleep {:?}", thread::current().id());
        //                 // sleep(Duration::from_secs(3)).await;
        //                 // println!("{:?} bytes received from {:?}", len, addr);
        //             });
        //             // PacketDispatcher::dispatcher(msg, addr, &peer);
        //             // peer.framed.send(msg.cover_send());
        //         }
        //         Err(e) => {
        //             tracing::error!(
        //                 "an error occurred while processing messages for; error = {:?}",
        //                 e
        //             );
        //         }
        //         _ => {}
        //     }
        // }
        // return;
    }
}

#[derive(Debug)]
pub struct UdpConnection<U: Decoder> {
    framed: SplitStream<UdpFramed<U>>,
    // rx: UnboundedReceiver<Message>,
    send: UnboundedSender<Message>,
}

impl<U: Decoder> UdpConnection<U> {
    pub fn new(
        framed: SplitStream<UdpFramed<U>>,
        state: Arc<Mutex<Shared>>,
        send: UnboundedSender<Message>,
    ) -> io::Result<UdpConnection<U>> {
        // Get the client socket address
        // let addr = framed.get_ref().peer_addr()?;

        // Create a channel for this peer
        // let (send, rx) = mpsc::unbounded_channel();

        // Add an entry for this `Peer` in the shared state map.
        // state.lock().await.peers.insert(addr, tx);
        // let x = framed.get_ref();
        Ok(UdpConnection {
            framed,
            // rx,
            send,
        })
    }
}

impl<U: Decoder + Decoder<Item = Packet>> Stream for UdpConnection<U> {
    type Item = Result<Message, DCodecError<U>>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // if let Poll::Ready(Some(v)) = Pin::new(&mut self.rx).poll_recv(cx) {
        //     //  根据不同的响应消息进行不同的处理
        //     return Poll::Ready(Some(Ok(v)));
        // }
        // 没有响应消息进行下一步接受消息
        // Secondly poll the `Framed` stream.
        // let result: Option<_> = futures::ready!(Pin::new(&mut self.framed).poll_next_unpin(cx));
        let result: Option<Result<(<U as Decoder>::Item, SocketAddr), <U as Decoder>::Error>> =
            futures::ready!(Pin::new(&mut self.framed).poll_next(cx));
        // let poll = self.framed.poll_next(cx);

        Poll::Ready(match result {
            // We've received a message we should broadcast to others.
            Some(Ok(message)) => Some(Ok(Message::Received(message.0, message.1))),

            // An error occurred.
            Some(Err(e)) => Some(Err(DCodecError::Io(e))),

            // The stream has been exhausted.
            None => None,
        })
    }
}


pub struct UdpProcessor{
    pub send: UnboundedSender<Message>,
    pub(crate) recv: Option<UnboundedReceiver<Message>>,
}
#[async_trait]
impl Processor for UdpProcessor {
    type Socket = UdpSocket;
    async fn poll(&mut self, socket: Self::Socket, service: Arc<PacketDispatcher>) ->Result<(), Box<dyn Error>>{
        let state = Arc::new(Mutex::new(Shared::new()));
        // let mut framed = UdpFramed::new(socket, BytesCodec::new());
        let framed = UdpFramed::new(socket, PacketBytesCodec::default());
        // let codec = *framed.codec();
        let (mut sink, mut stream) = framed.split();
        // let (send, mut recv) = mpsc::unbounded_channel();
        let (send, mut recv)=(self.send.clone(),self.recv.take().unwrap());
        // upd没有链接所以只有一个UdpConnection，所以这里分担分发工作
        tokio::spawn(async move {
            while let x1 = recv.recv().await.unwrap_or(Message::Close()) {
                match x1 {
                    Message::Send(packet, socket_addr) => {
                        sink.send((packet, socket_addr)).await;
                        sink.flush().await;
                        // sink.start_send((Packet,SocketAddr));
                    }
                    Message::Broadcast(_, _) => {}
                    Message::Forward(_, _) => {}
                    Message::Received(_, _) => {}
                    Message::Close() => {}
                }
            }
        });
        let mut udp_connection = UdpConnection::new(stream, state.clone(), send.clone()).unwrap();

        while let Some(result) = udp_connection.next().await {
            match result {
                // A message was received from a peer. Send it to the
                // current user.
                Ok(Message::Received(mut packet, addr)) => {
                    // peer.lines.send(&msg).await?;
                    let send = send.clone();
                    let service = service.clone();
                    tokio::spawn(async move {
                        println!(
                            "收到新链接:{:?},message:{:?},线程:{:?}",
                            addr,
                            packet,
                            thread::current().id()
                        );

                        let connection = Connection {
                            packet: packet,
                            send: send,
                            addr,
                        };
                        println!("收到新链接{:?}", connection);
                        let service = service.clone();
                        tokio::spawn(async move {
                            //     Dispatcher
                            service.dispatcher(connection);
                        });
                        // UdpServer::process(udp_connection,packet,addr).await;
                        // println!("sleep {:?}", thread::current().id());
                        // sleep(Duration::from_secs(3)).await;
                        // println!("{:?} bytes received from {:?}", len, addr);
                    });
                    // PacketDispatcher::dispatcher(msg, addr, &peer);
                    // peer.framed.send(msg.cover_send());
                }
                Err(e) => {
                    tracing::error!(
                        "an error occurred while processing messages for; error = {:?}",
                        e
                    );
                }
                _ => {}
            }
        }
        return Ok(());
    }
}